/**
 * 
 */
package cn.edu.nju.ws.sview.wyge.relations.index.indexTask;

import java.io.FileNotFoundException;
import java.io.IOException;
import java.sql.Connection;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Statement;
import java.util.ArrayList;
import java.util.LinkedList;
import java.util.Properties;
import java.util.Queue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;

import cn.edu.nju.ws.sview.database.DBConnectionFactory;
import cn.edu.nju.ws.sview.wyge.relations.datasets.DatasetItem;

/**
 * @author Weiyi Ge (geweiyi@gmail.com) 2010-12-15
 * 
 */
public class IndexTask {
	private int threads;
	private ArrayList<DatasetItem> dss;
	public static byte[] lock = new byte[0];
	private Queue<Integer> tasks = new LinkedList<Integer>();

	public IndexTask(ArrayList<DatasetItem> dss){
		try {
			this.dss = dss;
			getAllProperties();
			try {
				Connection conn = DBConnectionFactory.getConnection();//.getFalconConnection();
				Statement stmt = conn.createStatement();
				ResultSet rs = stmt
						.executeQuery("SELECT id FROM rdf_doc order by id");
				while (rs.next()) {
					tasks.add(rs.getInt(1));
				}
				rs.close();
				stmt.close();
				conn.close();
			} catch (SQLException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			}

			BlockingQueue<ArrayList<Integer>> q = new LinkedBlockingQueue<ArrayList<Integer>>(
					threads);
			Producer p = new Producer(q);
			new Thread(p).start();
			ArrayList<Thread> trds = new ArrayList<Thread>();
			for (int i = 0; i < threads; i++) {
				Consumer c = new Consumer(q);
				Thread ct = new Thread(c);
				trds.add(ct);
				ct.start();
			}

			boolean over = false;
			while (over == false) {
				over = true;
				for (Thread t : trds) {
					if (t.isAlive())
						over = false;
				}
				Thread.sleep(1000);
			}
		} catch (FileNotFoundException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		} catch (IOException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		} catch (InterruptedException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
	}

	/**
	 * Load properties
	 * 
	 * @throws FileNotFoundException
	 * @throws IOException
	 */
	private void getAllProperties() throws FileNotFoundException, IOException {
		Properties properties = new Properties();
		properties.load(DBConnectionFactory.class.getClassLoader()
				.getResourceAsStream("conf.properties"));
		threads = Integer.parseInt(properties.getProperty("index_thread_num"));
		properties.clear();
		properties = null;
	}

	class Producer implements Runnable {
		private final BlockingQueue<ArrayList<Integer>> queue;

		Producer(BlockingQueue<ArrayList<Integer>> q) {
			queue = q;
		}

		public void run() {
			try {
				while (true) {
					if (!tasks.isEmpty()) {
						queue.put(produce());
					} else {
						// add separators to terminal consumer
						for (int i = 0; i < threads; i++) {
							ArrayList<Integer> sep = new ArrayList<Integer>();
							queue.put(sep);
						}
						break;
					}
				}
			} catch (InterruptedException ex) {
				ex.printStackTrace();
			}
		}

		ArrayList<Integer> produce() {
			ArrayList<Integer> task = new ArrayList<Integer>();
			int j = 0;
			while (!tasks.isEmpty()) {
				task.add(tasks.poll());
				j++;
				if (j > 100000)
					break;
			}
			return task;
		}
	}

	class Consumer implements Runnable {
		private final BlockingQueue<ArrayList<Integer>> queue;

		Consumer(BlockingQueue<ArrayList<Integer>> q) {
			queue = q;
		}

		public void run() {
			try {
				while (true) {
					// check the separator to terminal consumer
					ArrayList<Integer> task = queue.take();
					if (task.size() == 0)
						break;
					consume(task);
				}
			} catch (InterruptedException ex) {
				ex.printStackTrace();
			} catch (Throwable e) {
				e.printStackTrace();
				System.exit(1);
			}
		}

		void consume(ArrayList<Integer> product) throws Throwable {
			new IndexTaskForEachThread(product, dss);
		}
	}
}
